package org.example.branch_tag;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * 案例介绍：
 * 1. 使用datagen生成测试数据
 * 2. 创建带有自动tag配置的Paimon表
 * 3. 将生成的数据插入Paimon表
 * 4. 启动流任务
 */
public class LogWriterAutoTag {
    public static void main(String[] args) throws Exception {
        // 配置Flink Web UI端口和并行度
        Configuration config = new Configuration();
        config.setString("rest.bind-port", "8082"); // 设置Web UI端口为8082
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(config);
        env.setParallelism(1); // 设置并行度为1
        env.enableCheckpointing(5000); // 设置checkpoint interval为5秒

        EnvironmentSettings settings = EnvironmentSettings.newInstance().inStreamingMode().build();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);

        // 创建Paimon目录
        String createCatalogSQL = "CREATE CATALOG paimon WITH (\n" +
                "    'type' = 'paimon',\n" +
                "    'warehouse' = 'file:///tmp/paimon'\n" +
                ");";
        System.out.println("正在执行SQL: \n" + createCatalogSQL);
        tableEnv.executeSql(createCatalogSQL);

        // 创建带有自动tag配置的Paimon表
        String createAutoTagTableSQL = "CREATE TABLE IF NOT EXISTS `paimon`.`default`.`AutoTagTable` (\n" +
                "    k INT PRIMARY KEY NOT ENFORCED,\n" +
                "    f0 INT,\n" +
                "    f1 STRING\n" +
                ") WITH (\n" +
                "    'connector' = 'paimon',\n" +
                "    'tag.automatic-creation' = 'process-time',\n" +  // 基于处理时间自动创建tag
                "    'tag.creation-period' = 'hourly',\n" +           // 每天创建一个tag
                "    'tag.creation-delay' = '0 m',\n" +             // 延迟10分钟创建tag
                "    'tag.num-retained-max' = '90'\n" +              // 最多保留90个tag
                ");";
        System.out.println("正在执行SQL: \n" + createAutoTagTableSQL);
        tableEnv.executeSql(createAutoTagTableSQL);

        // 使用datagen生成测试数据
        String createTempTableSQL = "CREATE TEMPORARY TABLE TempAutoTagTable (\n" +
                "    k INT,\n" +
                "    f0 INT,\n" +
                "    f1 STRING\n" +
                ") WITH (\n" +
                "    'connector' = 'datagen',\n" +
                "    'rows-per-second' = '5',\n" +
                "    'fields.k.min' = '1',\n" +
                "    'fields.k.max' = '1000',\n" +
                "    'fields.f0.min' = '0',\n" +
                "    'fields.f0.max' = '100',\n" +
                "    'fields.f1.length' = '10'\n" +
                ");";
        System.out.println("正在执行SQL: \n" + createTempTableSQL);
        tableEnv.executeSql(createTempTableSQL);

        // 将生成的测试数据插入Paimon表
        String insertSQL = "INSERT INTO `paimon`.`default`.`AutoTagTable`\n" +
                "SELECT\n" +
                "    k,\n" +
                "    f0,\n" +
                "    f1\n" +
                "FROM TempAutoTagTable;";
        System.out.println("正在执行SQL: \n" + insertSQL);
        tableEnv.executeSql(insertSQL);
    }
}
